package org.example.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.filter.SubstringComparator;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author zhl
 * @version 1.0
 * @ClassName WeiboHbase
 * @date 2020/11/7 13:37
 */
public class WeiboHbase {
    private static volatile WeiboHbase INSTANCE;
    private static Configuration configuration;                                     //hbase配置
    private static Connection connection;                                           //hbase connection
    private static final  byte[] CONTENT= Bytes.toBytes("weibo:content");        // 微博发表内容表存储
    private static final  byte[] RELATION=Bytes.toBytes("weibo:relationship");          // 微博关系存储表
    private static final  byte[] RECEIVE_CONTENT_EMAIL=Bytes.toBytes("weibo:receive-content"); // 用户订阅接收信息表
    private Table relationTB;
    private Table contentTable;
    private Table rceTable;
    private Admin admin;

    public static WeiboHbase getINSTANCE() {
        if (INSTANCE == null) {
            synchronized (WeiboHbase.class) {
                if (INSTANCE == null) {
                    INSTANCE = new WeiboHbase();
                }
            }
        }
        return INSTANCE;
    }

    public Admin getAdmin() {
        return admin;
    }

    public void setAdmin(Admin admin) {
        this.admin = admin;
    }

    public static Configuration getConfiguration() {
        return configuration;
    }

    public static void setConfiguration(Configuration configuration) {
        WeiboHbase.configuration = configuration;
    }

    public static Connection getConnection() {
        return connection;
    }

    public static void setConnection(Connection connection) {
        WeiboHbase.connection = connection;
    }

    private WeiboHbase() {
        try {
            if (configuration == null) {
                configuration = HBaseConfiguration.create();
                configuration.set("hbase.zookeeper.quorum", "mnode1:2181");
                connection = ConnectionFactory.createConnection(configuration);
                relationTB = connection.getTable(TableName.valueOf(RELATION));            // 用户关系表
                contentTable = connection.getTable(TableName.valueOf(CONTENT));           // 用户发表内容表
                rceTable = connection.getTable(TableName.valueOf(RECEIVE_CONTENT_EMAIL)); // 用户订阅内容表
                admin = connection.getAdmin();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 添加好友关注
     * */
    public void addAttends(String uid, String[] attends) {
        if (attends == null || attends.length <= 0) {
            return;
        }
        try {
            List<Put> puts = new ArrayList<>();     // 被关注对象增加粉丝列
            List<Put> rcePuts = new ArrayList<>();  // 给用户增加关注对象的订阅信息

            Scan scan = new Scan();
            // 在微博用户关系中 添加新的好友关注，微博邮件箱表添加关注用户发布的微博内容的rowKey
            Put attendsPut = new Put(uid.getBytes());
            for ( String attend : attends ){
                puts.add(addRelation(attendsPut, uid, attend));

                // 过滤出关注对象的信息
                RowFilter filter = new RowFilter(CompareOperator.EQUAL, new SubstringComparator(attend+"_"));
                scan.setFilter(filter);
                ResultScanner results = contentTable.getScanner(scan);
                rcePuts.addAll(addContent(results, uid, attend));
            }
            puts.add(attendsPut);
            relationTB.put(puts);
            rceTable.put(rcePuts);
        }catch (IOException e){
            e.printStackTrace();
        }
    }

    /**
     * 关闭连接
     * */
    public void close() {
        if (null != connection) {
            try {
                connection.close();
            }catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 增加被关注用户粉丝表
     * */
    private Put addRelation(Put attendsPut, String uid, String attend) {
        // 添加关注关系
        attendsPut.addColumn("attends".getBytes(), attend.getBytes(), attend.getBytes());
        // 添加被关注关系
        Put fansPut = new Put(attend.getBytes());
        fansPut.addColumn("fans".getBytes(), uid.getBytes(), uid.getBytes());
        return fansPut;
    }

    /**
     * 将关注对象发表的信息添加到用户订阅表中
     * */
    private List<Put> addContent(ResultScanner results, String uid, String attend) {
        Scan scan = new Scan();

        List<Put> resultPut = new ArrayList<>();
        RowFilter filter = new RowFilter(CompareOperator.EQUAL, new SubstringComparator(attend+"_"));
        scan.setFilter(filter);
        for (Result r : results) {
            for (Cell cell : r.listCells()) {
                byte[] rk = CellUtil.cloneRow(cell);
                Put put = new Put(Bytes.toBytes(uid));
                String rowKey = Bytes.toString(rk);

                long timestamp = Long.parseLong(rowKey.substring(rowKey.indexOf("_")+1));
                put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(attend),timestamp, rk);
                resultPut.add(put);
            }
        }
        return resultPut;
    }

    /**
     * 发布内容
     * @param uid 发布者id
     * @param content 发布内容
     * */
    public void publishContent(String uid, String content) {
        try {
            long timestamp = System.currentTimeMillis();
            String rowkey = uid + "_" + timestamp;      // 内容表的rowKey

            publish(rowkey, content, timestamp);        // 发布内容

            notifyFans(uid, timestamp, rowkey);   // 通知所有粉丝

        }catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 通知所有粉丝，在订阅表中添加内容信息
     * */
    private void notifyFans( String uid, long timestamp, String rowkey) throws IOException {
        List<byte[]> fans = getFans(uid);           // 获取用户粉丝

        List<Put> puts = new ArrayList<>();

        for (byte[] fan : fans) {
            Put fanPut = new Put(fan);
            fanPut.addColumn(Bytes.toBytes("cf"), Bytes.toBytes(uid), timestamp, Bytes.toBytes(rowkey));
            puts.add(fanPut);
        }
        rceTable.put(puts);
    }

    /**
     * 获取uid用户的所有粉丝
     * @param uid 需要被获取的用户的id
     * */
    private List<byte[]> getFans(String uid) throws IOException {
        Scan scan = new Scan();
        List<byte[]> fans = new ArrayList<>();
        ResultScanner rs = relationTB.getScanner(scan);
        RowFilter filter = new RowFilter(CompareOperator.EQUAL, new SubstringComparator(uid));
        scan.setFilter(filter);
        for (Result result : rs) {
            for (Cell cell : result.listCells()) {
                fans.add(CellUtil.cloneValue(cell));
            }
        }

        return fans;
    }

    /**
     * 获取用户所有粉丝的名称
     * @param uid 需要被获取的用户的id
     * */
    public List<String> getUserFans(String uid) throws IOException {
        List<String> fans = new ArrayList<>();
        for (byte[] fan : getFans(uid)) {
            fans.add(Bytes.toString(fan));
        }
        return fans;
    }

    /**
     * 发布内容，在内容表中添加字段
     * */
    private void publish(String rowkey, String content, long timestamp) throws IOException {
        Put put = new Put(Bytes.toBytes(rowkey));

        // column 内容
        put.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("content"), timestamp, Bytes.toBytes(content));
        // 添加column
        contentTable.put(put);
    }

    /**
     * 删除或取消关注用户
     * */
    public void removeAttends(String uid, String[] attends) {
        try {
            List<Delete> deletes = new ArrayList<>();
            Delete attendsDelete = new Delete(Bytes.toBytes(uid));
            Delete rceDelete = new Delete(Bytes.toBytes(uid));
            for (String attend : attends) {
                // 在用户的关注表中删除所有关注对象
                attendsDelete.addColumn(Bytes.toBytes("attends"), Bytes.toBytes(attend));

                // 从被关注用户角度来说，删除粉丝用户
                Delete fansDelete = new Delete(Bytes.toBytes(attend));
                fansDelete.addColumn(Bytes.toBytes("fans"), Bytes.toBytes(uid));
                deletes.add(fansDelete);

                // 在订阅信息表中，将该关注者订阅信息设置为无限长
                rceDelete.addColumns(Bytes.toBytes("cf"), Bytes.toBytes(attend));
            }

            deletes.add(attendsDelete);
            relationTB.delete(deletes);

            rceTable.delete(rceDelete);
        }catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取关注用户发布的微博内容
     * 创建表示设置cf版本为10
     * */
    public List<Content> getAttendsContents(String uid) {
        List<Content> contents = new ArrayList<>();
        try {
            List<byte[]> rowkeys =getRceAllCellsContent(uid);
            contents = getAllContentByRowKeys(rowkeys);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return contents;
    }

    /**
     * 获取指定用户信箱中的所有版本信息的值，就是content表中content的id
     * */
    private List<byte[]> getRceAllCellsContent(String uid) throws IOException {
        List<byte[]> rowkeys = new ArrayList<>();
        Get get = new Get(Bytes.toBytes(uid));
        get.readAllVersions();
        Result result = rceTable.get(get);
        for (Cell cell : result.rawCells()) {
            rowkeys.add(CellUtil.cloneValue(cell));
        }
        return rowkeys;
    }

    /**
     * 通过rowkey值获取所有的内容，并封装成Content类
     * */
    private List<Content> getAllContentByRowKeys(List<byte[]> rowkeys) throws IOException {
        Result[] results = getAllResultFromContentTBByRowKey(rowkeys);
        return getContentFromResults(results);
    }

    /**
     * 通过rowkeys获取所有Content表中的结果
     * */
    private Result[] getAllResultFromContentTBByRowKey(List<byte[]> rowkeys) throws IOException {
        List<Get> gets = new ArrayList<>();
        for (byte[] rk : rowkeys) {
            Get g = new Get(rk);
            gets.add(g);
        }
        return contentTable.get(gets);
    }

    /**
     * 将Result封装成content类的数组
     * */
    private List<Content> getContentFromResults(Result[] results) {
        List<Content> contents = new ArrayList<>();
        for (Result res : results) {
            for (Cell cell : res.rawCells()) {
                contents.add(getContentFromCell(cell));
            }
        }
        return contents;
    }

    /**
     * 将cell封装成content类
     * */
    private Content getContentFromCell(Cell cell) {
        Content content = new Content();
        String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
        String userId = rowKey.substring(0, rowKey.indexOf("_"));
        String timestamp = rowKey.substring(rowKey.indexOf("_")+1);
        String attendContent = Bytes.toString(CellUtil.cloneValue(cell));
        content.setUserId(userId);
        content.setTimestamp(timestamp);
        content.setContent(attendContent);
        return content;
    }

    public static void main(String[] args) throws IOException {
        WeiboHbase weiboHbase = WeiboHbase.getINSTANCE();
        weiboHbase.addAttends("fan2", new String[] {"user1"});
        weiboHbase.publishContent("user1", "test2");
//        for (byte[] b : weiboHbase.getFans("user1")) {
//            System.out.println(new String(b));
//        }
//        weiboHbase.getFans("user1");
        for (String s : weiboHbase.getUserFans("user1")) {
            System.out.println(s);
        }
        for (Content content : weiboHbase.getAttendsContents("zhl")) {
            System.out.println(content);
        }
//        weiboHbase.removeAttends("fan1", new String[] {"user1 "});
        weiboHbase.close();
    }
}
